<?php

class Pool
{
    protected $server;
    protected $min; //最小连接数
    protected $max; //最大连接数
    protected $idle_pool;//空闲连接池
    protected $worker_pool;//工作连接池
    protected $wait_queue;//等待队列
    protected $db_config; //配置信息
    protected $wait_queue_max;//最多的排队数
    protected $count; //记录当前数据库的连接数量
    protected $table;

    public function __construct($config)
    {
        //全局对象
        $this->server = new \swoole\server('0.0.0.0', 9501);
        $this->server->set($config['tcp']);
        //注册事件
        $this->server->on('WorkerStart', [$this, 'onWorkerStart']);
        $this->server->on('receive', [$this, 'onReceive']);
        $this->db_config = $config['db'];
        $this->max = $config['server']['max'];
        $this->min = $config['server']['min'];
        $this->idle_pool = new SplQueue();
        $this->wait_queue = new SplQueue();
        $this->wait_queue_max = 35;
        $this->table = new Swoole\Table(1024);
        $this->table->column('pool_count', Swoole\Table::TYPE_INT, 5);
        $this->table->create();
        $this->server->start(); //全局生命周期
    }

    public function onWorkerStart($server, $worker_id)
    {
        for ($i = 0; $i < $this->min; $i++) {
            $this->db_connect($this->db_config);
        }
        swoole_timer_tick(1000, function ($id) {
            $pool_count = $this->table->get(0, 'pool_count');
//            var_dump($this->wait_queue);
            var_dump('创建的连接数' . $pool_count . '----等待的队列数量' . count($this->wait_queue) . '----连接池中链接个数' . count($this->idle_pool));
        });
    }

    public function onReceive($server, $fd, $reactor_id, $data)
    {
        $len = unpack('N', $data)[1];
        $body = substr($data, -$len);//去除二进制数据之后,不要包头的数据
        $pool_count = $this->table->get(0, 'pool_count');
        //连接池中存在可用链接
        if (count($this->idle_pool) != 0) {
            $this->query($body, function ($res) {
//            var_dump($res);
            });
        } else {
            //连接池中可用链接不存在 重新建立数据库链接
//            var_dump('已经建立的链接数量'.$pool_count.'最大的链接数量'.$this->max);
//            echo $pool_count . PHP_EOL;
            if ($pool_count < ($this->max - $this->min)) {//连接池中连接数 小于最大的连接数 去创建链接 否者放入等待队列中
                $this->table->incr(0, 'pool_count', 1);
                $this->db_connect($this->db_config, function ($db) use ($body) {
//                    $this->push_queue('idle_pool', $db);
                    $this->query($body, function ($res) {
                        var_dump('重新建立链接', count($res));
                    });
                });
            } else {
                //等待队列中最多放入数量控制
                if (count($this->wait_queue) < $this->wait_queue_max) {
                    echo '加入队列' . PHP_EOL;
                    $this->push_queue('wait_queue', $body);

                }
            }

        }
//                    echo '等待的队列数量========' . count($this->wait_queue);
    }


    protected function query($sql, $callback = '')
    {
        $idle_pool = $this->idle_pool->shift();
        $db = $idle_pool['db'];
        $db->query($sql, function ($db, $res) use ($callback) {
//            sleep(1);
            if (is_callable($callback)) {
                $callback($res);
            }
            //执行完毕,重新放回到空闲连接池,复用连接
            $this->push_queue('idle_pool', $db);
            // //取出队列当中的
            if (count($this->wait_queue) > 0) {
                //空闲连接池当中去取出连接去执行sql
                $idle_num = count($this->idle_pool);
                for ($i = 0; $i < $idle_num; $i++) {
                    $req = $this->wait_queue->shift();
                    $this->query($req['db'], function ($res) {
                        //var_dump($res);
                        var_dump('等待队列的处理' . count($res));
                        //$this->server->send($fd,'123');
                    });
                }
            } else {

                $idle_num = count($this->idle_pool);
                var_dump('队列处理完毕' . $idle_num);
                for ($i = $idle_num; $i > 5; $i--) {
                    $idle_pool = $this->idle_pool->shift();
                    $idle_pool['db']->close();
                }
                $this->table->set(0, ['pool_count' => 0]);
            }
        });

    }

    /*
     * 异步数据库
     */
    public function db_connect($db_config, $callback = '')
    {
        $mysql = new Swoole\Mysql();
        $mysql->connect($db_config, function ($db, $res) use ($callback) {
            if ($res === false) {
                var_dump('链接失败：', $db->connect_errno, $db->connect_error);
                return;
            }
            $this->push_queue('idle_pool', $db);
//            $this->table->incr(0, 'pool_count', 1);
//            echo $this->table->get(0, 'pool_count').PHP_EOL;
            if (is_callable($callback)) {
                $callback($db);
                return;
            }
        });
    }

    //向队列当中添加连接
    public function push_queue($queueName, $pdo, $fd = 0)
    {
        $arr = [
            'fd' => $fd,
            'db' => $pdo
        ];
        $this->$queueName->push($arr);
    }


}

$config['tcp'] = [
    'worker_num' => 1,
//    'task_worker_num' => 20, //连接池数量
    'open_length_check' => true,
    'package_length_type' => 'N',
    'package_length_offset' => 0, //计算总长度
    'package_body_offset' => 4,//包体位置
];
$config['server'] = [
    'max' => 20,
    'min' => 5
];
$config['db'] = [
    'host' => '127.0.0.1',
    'port' => 3306,
    'user' => 'root',
    'password' => 'Abc123!Abc',
    'database' => 'test',
    'charset' => 'utf8', //指定字符集
    'timeout' => 10,  // 可选：连接超时时间（非查询超时时间），默认为SW_MYSQL_CONNECT_TIMEOUT（1.0）
];
new Pool($config);
